package com.hliushi.spark.sql

import org.apache.spark.sql.types._
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/18 15:31
 */
object MysqlWrite {

  /**
   * mysql的访问方式有两种: 使用本地运行, 提交到集群汇总运行
   * .
   * 写入MySQL数据时, 使用本地运行, 读取的时候使用集群运行
   *
   * 异常1: 这个是因为pom.xml 中没有添加 mysql连接的依赖
   * Exception in thread "main" java.sql.SQLException: No suitable driver
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    // 1.创建SparkSession对象
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("mysql write")
      .getOrCreate()


    // 2.读取数据创建DataFrame
    //  dataset/studenttab10k
    //    2.1.拷贝文件
    //    2.2.读取
    val schema = StructType(
      List(
        StructField("name", dataType = StringType),
        StructField("age", dataType = IntegerType),
        StructField("gpa", dataType = FloatType)
      )
    )

    val dataFrame = spark.read
      .schema(schema)
      .option("delimiter", "\t")
      .csv("dataset/studenttab10k")


    // 3.处理数据
    val resultDF = dataFrame.where("age < 30")

    // 4.落地数据
    // jdbc("jdbc:mysql://node03:3306/spark02", "stu", new Properties())
    /**
     * 函数原型
     * def jdbc(url: String, table: String, connectionProperties: Properties): Unit
     */
    resultDF.write
      .mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("url", "jdbc:mysql://node03:3306/spark02") // 驱动
      .option("dbtable", "student") // 数据库地址
      .option("user", "spark03") // 表名
      .option("password", "Spark03!")
      .save()

    spark.stop()
  }
}